package com.shuhe.thread;

import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.JsonNode;
import com.shuhe.common.BottomLevelTableGenerator;
import com.shuhe.common.MyDateUtils;
import com.shuhe.domain.alarm.TableSchema;
import com.shuhe.domain.alarm.TopicTable;
import com.shuhe.metastore.DBAppSql;
import org.apache.log4j.Logger;
import org.eclipse.paho.mqttv5.client.*;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;

import java.sql.Connection;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import static com.shuhe.common.CommonFunction.getUUID;
import static com.shuhe.thread.MqttReceiveThread.writeErrorMessage;

public class MqttPublishThread implements Runnable {
    private ArrayList<String> execs;

    private static Logger logger = Logger.getLogger(MqttPublishThread.class);

    private String brokerUrl;
    private String client_id;
    private String username;
    private String password;
    private ArrayList<TopicTable> topicTables;
    private Connection appConn;

    private String start_time = MyDateUtils.getNowDate();
    private String end_time = MyDateUtils.getNowDate();

    private MqttAsyncClient client = null;

    public MqttPublishThread(String brokerUrl, String client_id, String username, String password,
                             ArrayList<TopicTable> topicTables, Connection appConn) {
        this.brokerUrl = brokerUrl;
        this.client_id = client_id;
        this.username = username;
        this.password = password;
        this.topicTables = topicTables;
        this.appConn = appConn;
    }

    public void run() {
        try {
            client = new MqttAsyncClient(brokerUrl, client_id, new MemoryPersistence());
            MqttConnectionOptions options = new MqttConnectionOptions();
            options.setUserName(username);
            options.setPassword(password.getBytes());
            options.setCleanStart(true);
            options.setAutomaticReconnect(true); // 启用自动重连
            options.setKeepAliveInterval(60); // 设置心跳间隔（秒）
            options.setSessionExpiryInterval(0L);

            client.setCallback(new MqttCallback() {
                @Override
                public void disconnected(MqttDisconnectResponse disconnectResponse) {
                    logger.info("与 Broker 连接断开: " + disconnectResponse.getReasonString() + ": " + brokerUrl + ":" + client_id);
                }

                @Override
                public void mqttErrorOccurred(MqttException e) {
                    String errorMessage = "MQTT 错误: " + ": " + brokerUrl + ":" + client_id + e.getMessage();
                    writeErrorLog(errorMessage);
                    logger.error(errorMessage);
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                }

                @Override
                public void deliveryComplete(IMqttToken token) {
                    // 主要用于发布消息的确认，订阅方可不重点实现
                }

                @Override
                public void connectComplete(boolean reconnect, String serverURI) {
                    logger.info("连接成功: " + (reconnect ? "重连" : "首次连接") + ": " + brokerUrl + ":" + client_id);
                }

                @Override
                public void authPacketArrived(int reasonCode, MqttProperties properties) {
                    // 处理认证包，仅在需要特殊认证时使用
                }
            });

            // 连接到 Broker
            client.connect(options);
            int k = 0;
            while (true) {
                Thread.sleep(1000);
                k++;
                if (k > 60 * 12) {
                    client.disconnect();
                    client.close();
                    break;
                }

                for (int i = 0; i < topicTables.size(); i++) {
                    TopicTable topicTable = topicTables.get(i);
                    String startTime = MyDateUtils.getNowDate();
                    String table_name = topicTable.getDest_table_name();
                    String sql = "select * from " + table_name + " a\n" +
                            "WHERE NOT EXISTS (\n" +
                            "    SELECT 1 \n" +
                            "    FROM sys_alarminfo_publish_log l \n" +
                            "    WHERE l.sub_batch_id = CONCAT(a.aggregation_id, '_', a.aggregation_num)\n" +
                            ") ";
                    ResultSet rs = DBAppSql.querySql(appConn, sql);
                    while (rs.next()) {
                        JSONObject rowJson = new JSONObject();
                        java.sql.ResultSetMetaData metaData = rs.getMetaData();
                        int columnCount = metaData.getColumnCount();
                        for (int j = 1; j <= columnCount; j++) {
                            String columnName = metaData.getColumnName(j);
                            Object columnValue = rs.getObject(j);
                            rowJson.put(columnName, columnValue);
                        }
                        sendMessage(client, topicTable.getTopic(), rowJson.toJSONString());
                        String sub_batch_id = rs.getString("aggregation_id") + "_" + rs.getString("aggregation_num");
                        writeSuccessLog(sub_batch_id, topicTable, startTime, startTime, 1);
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            String errorMessage = "MQTT 错误: " + ": " + brokerUrl + ":" + client_id + e.getMessage();
            writeErrorLog(errorMessage);
            logger.error(errorMessage);
        } finally {
            if (client != null) {
                try {
                    client.disconnect();
                    client.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }

    }

    public static void sendMessage(MqttAsyncClient client, String topic, String payload)
            throws MqttException {

        // 创建 MQTT 5 消息
        MqttMessage message = new MqttMessage();
        message.setPayload(payload.getBytes());
        message.setQos(1);
        message.setRetained(false);

        // MQTT 5 属性
        MqttProperties properties = new MqttProperties();
        properties.setMessageExpiryInterval(3600L); // 消息过期时间（秒）
        message.setProperties(properties);

        // 发布消息
        IMqttToken pubToken = client.publish(topic, message);
        pubToken.waitForCompletion();
    }

    private void writeSuccessLog(String sub_batch_id, TopicTable topicTable, String start_time, String end_time,
                                 int data_count) {
        try {
            String sql = "insert into sys_alarminfo_publish_log(sub_batch_id, publish_id, status, error_msg, " +
                    "start_time, end_time, data_count) " +
                    "values (" + "'" + sub_batch_id + "','"
                    + topicTable.getSubscribe_id() + "','1', '', '" + start_time + "','" + end_time + "','"
                    + data_count + "')";
            DBAppSql.execSql(appConn, sql);
        } catch (Exception e) {
            System.err.println("写入日志失败: " + e.getMessage());
        }
    }

    private void writeErrorLog(String errorMessage) {
        try {
            String sql = "insert into sys_alarminfo_publish_log(sub_batch_id, publish_id, status, error_msg, " +
                    "start_time, end_time, data_count) " +
                    "values (" + "'" + getUUID() + "',"
                    + "'','0', '" + errorMessage + "', '" + MyDateUtils.getNowDate() + "','" + MyDateUtils.getNowDate()
                    + "',0)";
            DBAppSql.execSql(appConn, sql);

            writeErrorMessage(appConn, errorMessage, "msg_type_03");
        } catch (Exception e) {
            System.err.println("写入日志失败: " + e.getMessage());
        }
    }
}
